package engine

import (
	"fmt"
	"miniDW/src/datapack"
	"miniDW/src/interfaces"
	"miniDW/src/log"
	"sync"
)

const (
	COMPONENT_STOP    = 0
	COMPONENT_RUNNING = 1
)

type Component struct {
	name     string
	kind     string
	operator string
	provider string
	status   int

	sourceName        []string
	sourceTableStruct map[string]datapack.TableStruct
	sourceRowQueue    map[string]*datapack.RowQueue
	//sourceConnector 用于接收源组件的消息，为唯一接口
	sourceConnector *Connector

	targetName       []string
	table            *datapack.TableStruct
	targetConnectors map[string]*Connector
	function         interfaces.IComponentFunction
}

func (c *Component) PushRow(name string, row *datapack.Row) error {
	if _, ok := c.sourceRowQueue[name]; ok {
		c.sourceRowQueue[name].Enqueue(row)
		return nil
	} else {
		return log.Error(fmt.Errorf("no row queue for %s", name))
	}
}

func (c *Component) GetRowQueue(name string) *datapack.RowQueue {
	if _, ok := c.sourceRowQueue[name]; !ok {
		return nil
	}
	return c.sourceRowQueue[name]
}

func (c *Component) GetSourcesNames() []string {
	return c.sourceName
}

func (c *Component) GetSourceTableStruct(name string) (*datapack.TableStruct, error) {
	if tableStruct, ok := c.sourceTableStruct[name]; ok {
		return &tableStruct, nil
	}
	return nil, log.Error(fmt.Errorf("no such source component table struct: %s", name))
}

func (c *Component) SetSourceTableStruct(source string, table *datapack.TableStruct) error {
	if len(c.sourceTableStruct) == len(c.sourceName) {
		return log.Error(fmt.Errorf("source table struct length does not match source table length"))
	}
	if _, ok := c.sourceTableStruct[source]; ok {
		return log.Error(fmt.Errorf("source table struct already exists"))
	}
	c.sourceTableStruct[source] = *table

	//如果所有原的表机构设置完毕，则由组件自行处理
	if len(c.sourceTableStruct) == len(c.sourceName) {
		return c.function.SourceTableStructSetFinished(c)
	}
	return nil
}

func (c *Component) ProcessMsg(msg CommandMsg) error {
	switch msg.Cmd {
	case CMD_NOSOURCE:
		err := c.function.Process(c)
		if err != nil {
			return err
		}
	case CMD_STOP:
		c.status = COMPONENT_STOP
		_ = c.SendMessage(CMD_STOP, nil)
	case CMD_TABLESTRUCT:
		if tableStruct, ok := msg.Data.(*datapack.TableStruct); !ok {
			return log.Error(fmt.Errorf("Table struct not found in command"))
		} else {
			return c.SetSourceTableStruct(msg.Source, tableStruct)
		}
	case CMD_ROWDATA:
		if msg.Data == nil {
			fmt.Println("msg.Data is nil")
		}
		err := c.PushRow(msg.Source, msg.Data.(*datapack.Row))
		if err != nil {
			return err
		}
		err = c.function.Process(c)
		if err != nil {
			return err
		}
	}
	return nil

}

func (c *Component) Run(wg *sync.WaitGroup) {
	defer wg.Done()
	msg := CommandMsg{Cmd: CMD_NOSOURCE}
	c.status = COMPONENT_RUNNING

	for c.status == COMPONENT_RUNNING {
		msg.Cmd = CMD_NOSOURCE
		//如果有源,则从源头获取数据
		if c.sourceConnector != nil {
			msg = c.sourceConnector.Received()
		}
		err := c.ProcessMsg(msg)
		if err != nil {
			c.status = COMPONENT_STOP
			return
		}
	}
	log.Debug(fmt.Sprintf(`%s has been stopped`, c.name))
	return
}

func (c *Component) Open() error {
	err := c.function.Open(c)
	if err != nil {
		return err
	}
	return nil
}

func (c *Component) Close() error {
	if c.sourceConnector != nil {
		_ = c.sourceConnector.Close()
	}
	return nil
}

func (c *Component) Name() string {
	return c.name
}

func (c *Component) SourceName() *[]string {
	return &c.sourceName
}

func (c *Component) SetSourceName(names []string) {
	c.sourceName = names
	c.sourceRowQueue = make(map[string]*datapack.RowQueue, len(names))
	for _, name := range names {
		c.sourceRowQueue[name] = datapack.NewRowQueue()
	}
	return
}

func (c *Component) TargetName() *[]string {
	return &c.targetName
}

func (c *Component) SetTargetName(names []string) {
	c.targetName = names
	return
}

func (c *Component) GetSourceConnector() *Connector {
	if c.sourceConnector == nil {
		c.sourceConnector = NewConnector(c.name)
	}
	return c.sourceConnector
}

func (c *Component) AddTargetConnector(connector *Connector) error {
	if c.targetConnectors == nil {
		c.targetConnectors = make(map[string]*Connector)
	}
	name := connector.Target()
	if _, ok := c.targetConnectors[name]; ok {
		return fmt.Errorf("connector target %s already exists", name)
	}
	c.targetConnectors[name] = connector
	return nil
}

// 当数据为null时，视为数据结束
func (c *Component) SendRowData(row *datapack.Row) error {
	return c.SendMessage(CMD_ROWDATA, row)
}

func (c *Component) SendMessage(cmd int, data interface{}) error {
	msg := CommandMsg{
		Cmd:    cmd,
		Source: c.name,
		Data:   data,
	}
	for _, connector := range c.targetConnectors {
		connector.Send(msg)
	}
	return nil
}

func (c *Component) Stop() {
	c.status = COMPONENT_STOP
}

func (c *Component) GetTableStruct() *datapack.TableStruct {
	return c.table
}
func (c *Component) SetTableStruct(ts *datapack.TableStruct) error {
	c.table = ts
	err := c.SendMessage(CMD_TABLESTRUCT, ts)
	if err != nil {
		return err
	}
	return nil
}

func CreateComponent(cfg ConfigInfo) (*Component, error) {
	var err error
	component := &Component{}
	component.name = cfg.Name
	component.kind = cfg.Kind
	component.operator = cfg.Operator
	component.provider = cfg.Provider
	component.targetName = cfg.Targets
	component.table, err = datapack.NewTableStructByParam(cfg.Table)
	if err != nil {
		return nil, err
	}
	component.sourceTableStruct = make(map[string]datapack.TableStruct)
	if len(cfg.Sources) > 0 {
		component.SetSourceName(cfg.Sources)
	}

	function, err := GetInstance().CreateFunction(cfg.Kind, cfg.Operator, cfg.Provider, cfg.Params)
	if err != nil {
		return nil, err
	}
	component.function = function

	return component, nil
}
